package com.simple.datax.conn;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.SimpleFailMessage;
import com.simple.datax.SimpleMessage;
import com.simple.datax.SimpleObject;
import com.simple.datax.api.Connector;
import com.simple.datax.api.ConnectorException;
import com.simple.datax.conn.impl.SendThreadPoolExecutor;
import com.simple.datax.service.impl.SendStatusHandler;
import com.simple.datax.utils.MessageUtils;
import com.simple.datax.utils.SimpleConstants;

/** 
 * @Title: BaseConnector.java 
 * @Package com.simple.datax.model.connector 
 * @Description: 连接器抽象类，用于定义基本的发送和接收操作。
 * @author Jin Yi 
 * @date 2012-12-30 
 * @version V1.0
 */
public abstract class BaseConnector implements Connector {
	private static final Logger logger = LoggerFactory
			.getLogger(BaseConnector.class);
	// 当队列中没有文件时，超时时间
	protected static final int RECV_NO_MESSAGE_TIMOUT = 2000;
	
	//报文类型
	protected String msgType="EC";
	//报文格式
	protected String msgFormat="XML";
	//编码
	protected String encoding = "UTF-8";
	
	protected ConcurrentLinkedQueue<SimpleObject> messageList = new ConcurrentLinkedQueue<SimpleObject>();
	// Thead Safe
	protected ConcurrentLinkedQueue<SimpleObject> newMessageList = new ConcurrentLinkedQueue<SimpleObject>();
//	private Thread thread=null;
	
	public static final String TYPE = "BaseConnector";
	
	protected SendStatusHandler handler;
	
	protected String connPara;
	
	private static final char SPLITTER = '$';
	private static final char ESCAPE = '\\';
	
	protected SendExecutor executor;
	
	protected SendThreadPoolExecutor poolExecutor;
	
	private long total;
	private long successTotal;
	
	public String getEncoding() {
		return encoding;
	}

	public void setEncoding(String encoding) {
		this.encoding = encoding;
	}
	
	public String getMsgType() {
		return msgType;
	}

	public void setMsgType(String msgType) {
		this.msgType = msgType;
	}

	public String getMsgFormat() {
		return msgFormat;
	}

	public void setMsgFormat(String msgFormat) {
		this.msgFormat = msgFormat;
	}
	
	public BaseConnector(String connPara) throws ConnectorException {
//		executor = new SendExecutor(SendThreadPoolExecutor.getInstance(), connPara);
		if (connPara == null) {
			throw new ConnectorException("Connetion Parameter String is NULL. ");
		}
	}
	
	/**
	 * 外部处理程序需要发送消息是调用此函数。
	 * 此函数并不执行实际发送操作，而是放入连接器内容的队列中，等待连接器去执行发送操作。
	 * 
	 * @param object
	 */
	public boolean send(SimpleObject object) {
		if (object instanceof SimpleMessage) {
			// 将消息加入连接器消息队列中
			synchronized (newMessageList) {
				newMessageList.add(object);
				logger.debug("Call send - " + ((SimpleMessage)object).toLog());
//				logger.debug("===message cout in queue: " + newMessageList.size());
				if (poolExecutor == null) {
					poolExecutor = SendThreadPoolExecutor.getInstance();
				}
				executor = new SendExecutor(poolExecutor, connPara);
				poolExecutor.execute(executor, connPara);
//				logger.debug("Call poolExecutor - " + ((SimpleMessage)object).toLog());
//				try{
//					// 启动发送线程
//					startSendThread();
//				} catch (Exception ex) {
//					logger.error("Start sending thread failed", ex);
//				}
			}
			
		}
		return true;
	}
	
//	/**
//	 * 启动发送线程执行实际发送操作
//	 * 如果线程已经存在，则不启动新线程。
//	 */
////	private synchronized void startSendThread() {
//	private void startSendThread() {
//		/*
//		// If thread is running
//		if (thread != null && thread.isAlive()) {
//			// Do nothing
//		} else {
//			// start thread to go through message list
//			thread = null;
//			SendThread object = getThread();
//			thread = object.execute(executor);
////			thread = new Thread(new SendExecutor());
////			thread.start();
//		}
//		*/
//		if (poolExecutor == null) {
//			poolExecutor = SendThreadPoolExecutor.getInstance();
//		}
//		poolExecutor.execute(executor, connPara);
//	}
	
//	/**
//	 * 获取一个PoolableThread对象
//	 * 1) 首先，尝试通过PoolableThread Pool获取
//	 * 2) 如果没有获取到对象
//	 * 	2.1) 说明Pool已经达到最大Active数量，则需要记录日志，以提示系统达到饱和，需要扩容.
//	 *  2.2) 直接创建PoolableThread对象，以便系统能够继续执行发送操作。
//	 * @return
//	 */
//	private SendThread getThread() {
//		SendThread ret = null;
//		try {
//			ret = SendThreadPool.getInstance().borrowObject();
//		} catch (Exception e) {
//			logger.error("Creat thread to send failed.", e);
//		}
//		
//		if (ret == null) {
//			logger.error("Error: reach system capacity, please increase system capacity.");
//			try {
//				// 直接创建PoolableThread对象. Note: 用null作为 makeObject 参数，因为不需要还回此PoolableThread对象
//				ret = new SendThread(null);
//			} catch (Exception e) {
//				
//			}
//		}
//		return ret;
//	}
	
	/**
	 * @Description: 发送操作执行器
	 * @author Jin Yi
	 * @date 2012-12-30 
	 * @version V1.0
	 */
	private class SendExecutor implements Runnable {
		private SendThreadPoolExecutor executor;
		private String connPara;
		public SendExecutor(SendThreadPoolExecutor  executor, String connPara) {
			this.executor = executor;
			this.connPara = connPara;
		}
		
//		public String getKey() {
//			return connPara;
//		}
		
		@Override
		public void run() {
			synchronized (newMessageList) {
				messageList.addAll(newMessageList);
				newMessageList.clear();
			}
			
			while (true) {
//			while (newMessageList != null && newMessageList.size()>0) {
//				try {
				logger.debug(" start executor run:");
				SimpleMessage src = null;
				try {
					prepareSend();
				} catch (Exception e) {
					//如果是网络问题，不做任何处理，这样系统将在以后重试发送。其他异常，系统将标记报文处理失败
					logger.error("Send Connector initialization failed.", e);
//					logger.error(" message size:"+messageList.size());
					messageList.clear();
					executor.workDone(connPara);
					break;
				}
				for (SimpleObject object:messageList) {
					try {
						src = ((SimpleMessage)object);
						total++;
						internalSend(src);
						successTotal++;
						logger.debug("Sent successed - " + src.toLog());
						if (handler == null) {
							handler = SendStatusHandler.getInstance();
						}
						
						handler.callSuccessService(src);
					} catch (ConnectorException ex) {
						// 如果是网络问题，不做任何处理，这样系统将在以后重试发送。其他异常，系统将标记报文处理失败
						if (!ex.isNetworkIssue()) {
							// 断开已有连接，重新键连接，并重试发送
							handleError(ex, src);
							logger.error("Failed to send message caused by other issue." + src.toLog(), ex);
						} else {
							logger.debug("Failed to send message casued by network issue. Will re-init channel and try again." + src.toLog(), ex);
							try {
								disconnect();
								prepareSend();
								internalSend(src);
							} catch (ConnectorException e) {
								if (!e.isNetworkIssue()) {
									handleError(ex, src);
								} else {
									logger.warn("Wait to re-send - " + src.toLog(), ex);
								}
							} catch (Exception e) {
								handleError(e, src);
							}
						}
					} catch (Exception ex) {
						handleError(ex, src);
					}
				}
//				} catch (Exception ex) {
//					logger.error("Fatal error happened in sending message.", ex);
//					break;
//				}
				messageList.clear();
				synchronized (newMessageList) {
					if (newMessageList == null || newMessageList.size()==0) {
						executor.workDone(connPara);
						break;
					} else {
						messageList.addAll(newMessageList);
						newMessageList.clear();
					}
				}
			}
		}
	}
	
	/**
	 * 为接收消息做准备。所有 Connector 必须实现各自的初始化和连接有效等检查
	 * @throws ConnectorException
	 */
	abstract protected void prepareReceive() throws ConnectorException;
	
	/**
	 * 发送消息，所有 Connector 必须实现各自的发送方法
	 * @param message
	 * @throws ConnectorException
	 */
	abstract protected void internalSend(SimpleMessage message) throws ConnectorException;

	/**
	 * 为发送消息做准备。所有 Connector 必须实现各自的初始化和连接有效等检查
	 * @throws ConnectorException
	 */
	abstract protected void prepareSend() throws ConnectorException;
	

	public abstract String getConnType();

	public String getConnPara() {
		return connPara;
	}

	public void setConnPara(String connPara) {
		this.connPara = connPara;
	}

	public SendStatusHandler getHandler() {
		return handler;
	}

	public void setHandler(SendStatusHandler handler) {
		this.handler = handler;
	}
	
	private void handleError(Exception ex, SimpleMessage src) {
		logger.error("Fail to send out message.", ex);
		String errorinfo = ex.getMessage();

		try {
			// 失败处理
			SimpleFailMessage sfm = MessageUtils.castFailMessage(
					errorinfo, SimpleConstants.ERROR_TYPE_SEND,
					SimpleConstants.ERROR_LEVEL_NORMAL,
					src);

			handler.callFailService(sfm);// 注意这里需要处理SimpleFailMessage
		} catch (Exception e) {
			logger.error("Sender FailMessage Error:", e);
		}

		try {
			// 处理错误response报文
			int responseCode = SimpleConstants.RESPONSE_CODE_SENDER;
			String responseMsg = SimpleConstants.RESPONSE_MSG_SENDER;
			handler.callResponseService(src, responseCode,
					responseMsg, src.isNeedResponse());
		} catch (Exception e) {
			logger.error("Sender FailMessage Response Error:", e);
		}
	}
	
	/**
	 * 静态方法，用于解析参数串
	 * @param connPara
	 * @return String[] 参数数组
	 */
	public static String[] splitConnPara(String connPara) throws ConnectorException {
		if (connPara==null) {
			throw new ConnectorException("Connection Parameter String is NULL. ");
		}
		List<String> retList = new ArrayList<String>();
		char b = '\0';
		StringBuilder sb = new StringBuilder();
		int duplicate = 1;
		int length = connPara.length();
		if (length==0) {
			return new String[]{""};
		}
		for (int i=0; i<length; i++) {
			b = connPara.charAt(i);
			
			// 如果为分隔符，则将sb中保存的那内容移动到List中，然后清除sb中的内容。
			if (b == SPLITTER) {
				retList.add(sb.toString());
				sb.delete(0, sb.length());
			} else if (b == ESCAPE) { // 如果为转义符
				// next char should be ESCAPE too
				// 连续出现的SPLITTER的个数
				duplicate = 1;
				
				while ((i<length-1) && connPara.charAt(i+1)==ESCAPE) {
					duplicate++;
					i++;
				}
				
				// 连续出现的ESCAPE，实际含义为 一半个数的ESCAPE字符
				for(int j=0; j<duplicate/2; j++) {
					sb.append(ESCAPE);
				}
				
				// 剩余单个ESCAPE，应该是为转义$所用
				if (duplicate % 2==1) {
					if (i<length-1) {
						if (connPara.charAt(i+1)==SPLITTER) {
							sb.append(SPLITTER);
							i++;
						} else {
							throw new ConnectorException("Escape char \"\\\" needs double if it is used as normal char. ");
						}
					} else {
						throw new ConnectorException("Escape char \"\\\" needs double if it is used as normal char. ");
					}
				} else {
					// Do nothing
				}
				
			} else {
				sb.append(b);
			}
		}
		
		// 尾部剩余的部分
		if (sb.length() > 0) {
			retList.add(sb.toString());
			sb.delete(0, sb.length());
		}
		// 如果最后一个字符为$，则表示还有一个参数（虽然为空）
		if (b == SPLITTER) {
			retList.add(sb.toString());
		}
		return retList.toArray(new String[retList.size()]);
	}

	public SendThreadPoolExecutor getPoolExecutor() {
		return poolExecutor;
	}

	public void setPoolExecutor(SendThreadPoolExecutor poolExecutor) {
		this.poolExecutor = poolExecutor;
	}
	
	public long getTotal() {
		return total;
	}
	
	public long getSuccessTotal() {
		return successTotal;
	}
	
	/**
	 * 将字符串转换成指定编码的字节数组。<br>
	 * Example:<br>
	 * "1112"	-&gt; 494950 (GBK)
	 * 			-&gt; 494950 (UTF-8)
	 * 			-&gt; 490490500 (UTF-16)
	 * @param str
	 * @param encoding
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	public static byte[] getOutputBytes(String str, String encoding) throws UnsupportedEncodingException {
		if (str == null || str.trim().isEmpty()) {
			return null;
		}
		return str.getBytes(encoding);
	}
	
	/**
	 * 将字节数组转为字符串。字节数组的编码为encoding。<br>
	 * @param bytes
	 * @param encoding
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	public static String getInputString(byte[] bytes, String encoding) throws UnsupportedEncodingException {
		if (bytes == null || bytes.length==0) {
			return null;
		}
		return new String(bytes, encoding);
	}
	
	/**
	 * 将系统默认编码字符串转换为指定编码的字符串
	 * @param str
	 * @param encoding
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	public static String getOutputString(String str, String encoding) throws UnsupportedEncodingException {
		if (str == null || str.trim().isEmpty()) {
			return str;
		}
		
		return new String(str.getBytes(), encoding);
	}
	
	/**
	 * 将指定编码的字符串转为系统默认编码的字符串
	 * @param str
	 * @param encoding
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	public static String getInputString(String str, String encoding) throws UnsupportedEncodingException {
		if (str == null || str.trim().isEmpty()) {
			return str;
		}
		
		return new String(str.getBytes(encoding));
	}
}
